package com.atguigu.state;

import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by RUI on 2021/6/16 20:50
 */
public class State_Broadcast {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<String> dataStream = env.socketTextStream("hadoop162", 9999);

        MapStateDescriptor<String, String> controlState = new MapStateDescriptor<>("controlState", String.class, String.class);
        BroadcastStream<String> controlStream = env.socketTextStream("hadoop162", 8888).broadcast(controlState);

        dataStream.connect(controlStream)
                .process(new BroadcastProcessFunction<String, String, String>() {
                    @Override
                    public void processElement(String value, ReadOnlyContext ctx, Collector<String> out) throws Exception {
                        String aSwitch = ctx.getBroadcastState(controlState).get("switch");
                        if ("a".equals(aSwitch) || "b".equals(aSwitch)) {
                            System.out.println("切换至：" + aSwitch);
                        }else {
                            System.out.println("do nothing");
                        }
                    }

                    @Override
                    public void processBroadcastElement(String value, Context ctx, Collector<String> out) throws Exception {
                        // 处理广播元素: 把收到的值, 放入广播状态. 处理处理逻辑中就可以拿到这个广播状态
                        ctx.getBroadcastState(controlState).put("switch",value);
                    }
                })
                .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
